package com.dahua.messagesyncbj.runner;

import cn.hutool.core.convert.Convert;
import com.alibaba.fastjson2.JSON;
import com.dahua.messagesyncbj.config.redis.RedisUtil;
import com.dahua.messagesyncbj.criteria.SensorQueryCriteria;
import com.dahua.messagesyncbj.entity.Sensor;
import com.dahua.messagesyncbj.entity.Status;
import com.dahua.messagesyncbj.criteria.StatusQueryCriteria;
import com.dahua.messagesyncbj.pojo.Body;
import com.dahua.messagesyncbj.pojo.EquipmentInfo;
import com.dahua.messagesyncbj.pojo.MeasureInfo;
import com.dahua.messagesyncbj.service.EquipmentService;
import com.dahua.messagesyncbj.service.FaultService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import static com.dahua.messagesyncbj.business.Constants.SENSOR_HASH;

@Slf4j
@Component
public class KafkaHandleRunner implements Runnable {

    @Resource
    private RedisUtil redisUtil;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private EquipmentService equipmentService;
    @Resource
    private FaultService faultService;

    private List<ConsumerRecord<String, String>> consumerRecords;

    public KafkaHandleRunner(RedisUtil redisUtil,
                             RedissonClient redissonClient,
                             EquipmentService equipmentService,
                             FaultService faultService,
                             List<ConsumerRecord<String, String>> consumerRecords) {
        this.redisUtil = redisUtil;
        this.redissonClient = redissonClient;
        this.equipmentService = equipmentService;
        this.faultService = faultService;
        this.consumerRecords = consumerRecords;
    }


    private static final List<String> NO_PID_TYPE = Collections.singletonList("BHFT");

    @Override
    public void run() {
        this.handle(consumerRecords);
    }

    public void  handle(List<ConsumerRecord<String, String>> consumerRecords) {
        for (ConsumerRecord<String, String> item : consumerRecords) {
            try {
                if (item.value() == null || item.value().isEmpty()) {
                    continue;
                }
                Body body = JSON.parseObject(item.value(), MeasureInfo.class).getBody();
                RLock lock = redissonClient.getLock(body.getMacid() + "." + body.getPid());
                try {
                    lock.lock();
                    Body res = JSON.parseObject(JSON.toJSONString(redisUtil.getHash(body.getMacid(), body.getPid())), Body.class);
                    if (res == null || res.getTime() <= body.getTime()) {
                        redisUtil.setHash(body.getMacid(), body.getPid(), body);
                    }
                } finally {
                    try {
                        lock.unlock();
                    } catch (Exception e) {
                        log.error("【exception 1】：{}",e.getMessage());
                    }

                }
                this.setLatestStatus(body);

            } catch (Exception e) {
                log.error("【exception 2】：{}",e.getMessage() + item.value());
            }
        }
    }

    public void setLatestStatus(Body body) {

        EquipmentInfo info = null;
        info = equipmentService.enquire(body.getMacid());
        if (info == null) {
            return;
        }
        if (info.getBk82() == null){
            log.error("【设备未设置测点类型】：{}", body.getMacid());
            return;
        }

        Sensor sensor = null;
        String pid = body.getPid();


        /*  只通过bk82获取测点*/
        SensorQueryCriteria criteria = SensorQueryCriteria.builder()
                .bk1(info.getBk82())
                .isStatus(1)
                .build();
        List<Sensor> list = equipmentService.listSensors(criteria);
        if (list == null || list.isEmpty()){
            return;
        }

        sensor = list.get(0);
        boolean isNoPidType = NO_PID_TYPE.contains(info.getBk2());
        if (isNoPidType && !(body.getPid().equals(sensor.getPointDesc()))){
            return;
        }
        if ( !isNoPidType && !(body.getMacid() + sensor.getPid()) .equals(pid)){
            return;
        }


        /*  这里需要根据设备名前缀获取测点集。该方法可能会受到相同设备名前缀，但是不属于一类设备测点的错误*/

        /*String macType = body.getMacid().replaceAll("\\s*", "")
                .replaceAll("[^(a-zA-Z)]", "");
        Map<String, List<Sensor>> map = (Map<String, List<Sensor>>)JSON.parseObject(JSON.toJSONString(redisUtil.getHash(SENSOR_HASH, macType)), Map.class);

        if (map == null || map.isEmpty()) {
            //更新sensor info
            RLock lock = redissonClient.getLock("reset_sensors_info");
            try {
                lock.lock();
                equipmentService.resetSensorInfo();
            } catch (Exception e) {
                log.warn(e.getMessage());
            } finally {
                lock.unlock();
            }
            return;
        }

        if (map.containsKey("ALL")) {
            List<Sensor> list = map.get("ALL");
            if (list == null) {
                return;
            } else {
                boolean is_exist = false;
                for (Object o : list) {
                    Sensor sensor_t = JSON.parseObject(JSON.toJSONString(o), Sensor.class);
                    String t_pid = body.getMacid() + sensor_t.getPid();
                    if (t_pid.equals(pid)) {
                        is_exist = true;
                        sensor = sensor_t;
                    }
                }
                if (!is_exist) {
                    return;
                }
            }
        } else {
            List<Sensor> list = map.get(info.getBk82());
            if (list == null) {
                return;
            }
            boolean is_exist = false;
            for (Object o : list) {
                Sensor sensor_t = JSON.parseObject(JSON.toJSONString(o), Sensor.class);
                String t_pid = body.getMacid() + sensor_t.getPid();
                *//*  增加系统分类*//*
                if (t_pid.equals(pid) && sensor_t.getBk1().equals(info.getBk82())) {
                    is_exist = true;
                    sensor = sensor_t;
                }
            }
            if (!is_exist) {
                return;
            }
        }*/


        Status status = new Status();
        status.setLineId(info.getLineId());
        status.setStationId(info.getStationId());
        status.setEquipmentId(info.getId());
        status.setManufactureId(info.getName());
        status.setEquipmentType(info.getBk2());
        status.setLogTime(new Timestamp(body.getTime()));
        status.setStatusPoint(body.getPid());
        RLock lock = redissonClient.getLock("status." + info.getName());
        try {

            lock.lock();
            List<Status> statuses = equipmentService.listLatestStatus(new StatusQueryCriteria(info.getName()));
            if (statuses.isEmpty()) {
                this.setStatus(status, body, sensor, info.getBk2(), info);
                equipmentService.insertLatestStatus(status);
            } else {
                Status history = statuses.get(0);
                status.setStatus(history.getStatus());
                status.setSensorType(history.getSensorType());

                this.setStatus(status, body, sensor, info.getBk2(), info);
                if (history.getLogTime().before(status.getLogTime())) {
                    equipmentService.updateLatestStatus(status);
                }

            }
        } catch (Exception e) {
            log.error("【exception 3】：{}",e.getMessage());
        } finally {
            lock.unlock();
        }
    }

    /*  dahua标准：0-正常，1-预警，2-告警，3-报警，4-断网，5-停止*/

    /*  classify fault into five levels, with lower level being more imperative to fix, in order to obey beijing criteria*/
    /*  0:normal, 1-5:alert, 6-cut off, 7-stop*/
    public void setStatus(Status status, Body body, Sensor sensor, String type, EquipmentInfo info) {
        /*  这两类设备开头都是ZFJ,直接判断类型，后续要更正*/
        if (info.getBk2().equals("ZSFJ") || info.getBk2().equals("ZPFJ")) {
            switch (Convert.toInt(body.getValue())) {
                case -2: {
                    status.setStatus(6);
                    status.setSensorType(null);
                    break;
                }
                case -1: {
                    status.setStatus(7);
                    status.setSensorType(null);
                    break;
                }
                case 0: {
                    if (status.getStatus() > 0 && status.getStatus() <= 5) {
                        faultService.fixAllFault(status.getManufactureId(), body.getTime());
                    }
                    status.setStatus(0);
                    status.setSensorType(null);
                    break;
                }
            }
        }
        //对接主流风机:-2断网 -1停机  0正常 1报警 2预警 3故障
        else if (type.equals("ZFJ") || type.equals("SLFJ")) {
            switch (Convert.toInt(body.getValue())) {
                case -2: {
                    status.setStatus(6);
                    status.setSensorType(null);
                    break;
                }
                case -1: {
                    status.setStatus(7);
                    status.setSensorType(null);
                    break;
                }
                case 0: {
                    if (status.getStatus() > 0 && status.getStatus() <= 5) {
                        faultService.fixAllFault(status.getManufactureId(), body.getTime());
                    }
                    status.setStatus(0);
                    status.setSensorType(null);
                }
            }
        }
        //对接水泵：-1-停止，0-正常，1-故障
        else if (type.equals("SB")) {
            switch (Convert.toInt(body.getValue())) {
                case -1: {
                    status.setStatus(7);
                    status.setSensorType(null);
                    break;
                }
                case 0: {
                    if (status.getStatus() > 0 && status.getStatus() <= 5) {
                        faultService.fixAllFault(status.getManufactureId(), body.getTime());
                    }
                    status.setStatus(0);
                }
            }
        }
        //对接泵房：-1-自动，0-手动
        else if (type.equals("BF")) {
            switch (Convert.toInt(body.getValue())) {
                case 0: {
                    status.setBk1("转换开关为手动状态");
                    break;
                }
                case 1: {
                    status.setBk1("转换开关为自动状态");
                    break;
                }
            }
        }
        //对接通风空调
        else if (type.equals("TFKT")) {


        }
        //对接蓄电池
        else if (type.equals("XDC")) {


        }
        else if (type.equals("EPS")) {


        }
        else if (type.equals("UPS")) {


        }
        //对接博华扶梯
        else if (type.equals("BHFT")) {
            status.setStatus(Convert.toInt(body.getValue()));
            log.info("收到博华扶梯{}数据", info.getName());
        }
    }
}
